package com.eascs.spider.yuqing.schdule.dispatcher;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.http.impl.client.CloseableHttpClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.eascs.spider.yuqing.common.proj.Task;
import com.eascs.spider.yuqing.common.util.HttpUtil;
import com.eascs.spider.yuqing.common.util.PropertiesUtil;
import com.eascs.spider.yuqing.schdule.bloomfilter.BloomFilter;
import com.eascs.spider.yuqing.schdule.register.RegisterCentre;

/**
 * 任务分发
 * @author hzp
 *
 */
@Service
public class Dispatcher {

	private Queue<Task> queue;//任务优先级队列
	private static String workerRecievePath = PropertiesUtil.getConfig("worker.receive.path");
	
	@Autowired
	BloomFilter bf;

	@Autowired
	RegisterCentre registerCentre;
	
	Dispatcher() {
		 queue = new ConcurrentLinkedQueue<Task>();
		 CloseableHttpClient client = HttpUtil.initHttpClient();
		 Executors.newScheduledThreadPool(1).scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					if(!queue.isEmpty()){
						String workerPath = registerCentre.getRandomWorkerAddress();
						HttpUtil.executePost(client, "http://" + workerPath + workerRecievePath, queue.poll());
					}
				}
			 } ,100 ,50 ,TimeUnit.MILLISECONDS);
	}
	
	public synchronized void addTask(Task task){
		if(!bf.contains(task.getUrl())){
			queue.add(task);
			bf.add(task.getUrl());
		}
	}
	
	/**
	 * 添加首页请求任务,不需去重
	 * @param task
	 */
	public void addHomepageTask(Task task){
		queue.add(task);
	}
	
	public int taskSize(){
		return queue.size();
	}
}